package com.metriql.warehouse.metriql

import io.trino.connector.system.SystemHandleResolver
import io.trino.connector.system.SystemPageSourceProvider
import io.trino.connector.system.SystemSplitManager
import io.trino.connector.system.SystemTablesMetadata
import io.trino.connector.system.SystemTransactionHandle
import io.trino.metadata.InternalNodeManager
import io.trino.spi.connector.Connector
import io.trino.spi.connector.ConnectorContext
import io.trino.spi.connector.ConnectorFactory
import io.trino.spi.connector.ConnectorHandleResolver
import io.trino.spi.connector.ConnectorPageSourceProvider
import io.trino.spi.connector.ConnectorSplitManager
import io.trino.spi.connector.ConnectorTransactionHandle
import io.trino.spi.transaction.IsolationLevel
import io.trino.transaction.InternalConnector
import io.trino.transaction.TransactionId

class ExternalConnectorFactory(private val internalNodeManager: InternalNodeManager, private val name: String, val catalog: CatalogFile.Catalogs.Catalog) : ConnectorFactory {
    override fun getName() = name

    override fun create(catalogName: String?, config: MutableMap<String, String>?, context: ConnectorContext?): Connector {
        return ExternalConnector(internalNodeManager, catalog)
    }

    override fun getHandleResolver(): ConnectorHandleResolver {
        return SystemHandleResolver()
    }

    class ExternalConnector(private val nodeManager: InternalNodeManager, val catalog: CatalogFile.Catalogs.Catalog) : InternalConnector {

        override fun beginTransaction(transactionId: TransactionId, isolationLevel: IsolationLevel?, readOnly: Boolean): ConnectorTransactionHandle {
            return MetriqlTransactionHandle(transactionId)
        }

        override fun getSplitManager(): ConnectorSplitManager {
            return SystemSplitManager(nodeManager, catalog.value)
        }

        override fun beginTransaction(isolationLevel: IsolationLevel?, readOnly: Boolean): ConnectorTransactionHandle {
            return super.beginTransaction(isolationLevel, readOnly)
        }

        override fun getPageSourceProvider(): ConnectorPageSourceProvider {
            return SystemPageSourceProvider(catalog.value)
        }

        override fun getMetadata(transactionHandle: ConnectorTransactionHandle?) = SystemTablesMetadata(catalog.value)
    }

    class MetriqlTransactionHandle(transactionId: TransactionId?) : SystemTransactionHandle(transactionId, object : ConnectorTransactionHandle {})
}
